package com.ruoyi.flink.wc.util;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import com.ruoyi.flink.util.db.ClickHouseLink;
import com.ruoyi.flink.wc.entity.Participles;

public class ParticiplesRead extends RichSourceFunction<Participles>{

	private static final long serialVersionUID = 1L;
	Connection connection = null;
    PreparedStatement preparedStatement = null;
    String sql;
    public ParticiplesRead(String sql) {
        this.sql = sql;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = ClickHouseLink.getConn();
    }

    
	@Override
	public void run(SourceContext<Participles> ctx) throws Exception {
		  PreparedStatement preparedStatement = connection.prepareStatement(sql);
	        ResultSet resultSet = preparedStatement.executeQuery();
	        if(resultSet!=null) {
	            while(resultSet.next()) {
	            	ctx.collect(new Participles(resultSet.getString("words"),resultSet.getLong("num"),resultSet.getLong("timestamp")));
	            }
	        }
		
	}

	@Override
	public void cancel() {
        try {
            if(preparedStatement != null) {
                preparedStatement.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }

		
	}

}
